package com.baishancloud.log.quality

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.quality.pojo._
import com.baishancloud.log.quality.sink.HttpPostSink
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, WindowedStream}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import java.time.Duration
import java.util
import scala.collection.mutable

class MobileTerminal extends Serializable

/**
 *
 * @author ziqiang.wang 
 * @date 2021/11/15 19:03
 */
object MobileTerminal extends Serializable {


  /**
   * 处理移动端的数据，包括数据转化、聚合、计算、输出
   *
   * @param parameterTool 参数对象
   * @param business      业务：直播或点播：live、vod
   * @param stream        数据源
   */
  def mobileTerminal(parameterTool: ParameterTool, business: String, stream: DataStream[String]): Unit = {
    val parse: DataStream[(MobileLog, StarRocksMobile)] = stream
      .map(new MobileLogParse(business)).name("tuple2").uid("1e064645-0a31-4a02-9fb8-7e929c1e556c" + business)

    //明细数据入starRocks
    parse
      .map(_._2).name("StarRocksMobile").uid("4b19e8cb-35e4-4c55-a6ac-1181b2c92c5a")
      .filter(_ != null).name("!=null").uid("88ab9c44-b85a-4b11-9304-c02294758f36")
      .map(JSONUtil.toJsonStr(_)).name("toJsonStr").uid("6242ece2-57f5-4305-b862-e79c17523b0f")
      .addSink(SinkUtil.starRocksJsonString(parameterTool)).setParallelism(parameterTool.getInt(sinkDorisParallel, 1)).name("starRocks").uid("026968ec-99f9-47c9-aac2-e98aec659866")


    //比率计算
    val windowStream: WindowedStream[MobileLog, Tags, TimeWindow] =
      parse
        .map(_._1).name("MobileLog").uid("c54318c6-4ada-4db1-86e5-489d4b776d24")
        .filter(_ != null).name("!=null").uid("5f48f293-110b-4b5b-a3fc-bf70233b228a" + business)
        .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMinutes(parameterTool.getLong(maxOutOfOrdernessMinutes, 0))).withTimestampAssigner(new SerializableTimestampAssigner[MobileLog] {
          override def extractTimestamp(element: MobileLog, recordTimestamp: Long): Long = element.timestamp
        }))
        .keyBy(_.tags)
        .window(TumblingEventTimeWindows.of(Time.minutes(parameterTool.getLong(aggMinute, 5))))

    val timeCardFrameRateResult: DataStream[RateResult] = timeCardFrameRateCalculate(windowStream, business)
    val vvCardFrameRateResult: DataStream[RateResult] = vvCardFrameRateCalculate(windowStream, business)
    val loadFailedRateResult: DataStream[RateResult] = loadFailedRateCalculate(windowStream, business)
    timeCardFrameRateResult
      .union(vvCardFrameRateResult, loadFailedRateResult)
      .addSink(new HttpPostSink(parameterTool.getInt(sinkPostThreshold, Int.MaxValue))).setParallelism(1).name(s"移动${business}比率").uid(s"4f0c1afb-5848-4129-a30d-9ced0bd75bd1${business}")
  }


  /**
   * 时长卡顿率<br>
   * sum(if(asset_item_time_of_session<18000000 and first_video_time > 0 ,buffering_count,0))/ <br>
   * sum(if(asset_item_time_of_session<18000000 and first_video_time > 0 and asset_item_time_of_session >0, asset_item_time_of_session/ 60000,0))
   *
   */
  private def timeCardFrameRateCalculate(windowStream: WindowedStream[MobileLog, Tags, TimeWindow], business: String = ""): DataStream[RateResult] = {
    windowStream.process(new ProcessWindowFunction[MobileLog, RateResult, Tags, TimeWindow] {
      override def process(key: Tags, context: Context, elements: Iterable[MobileLog], out: Collector[RateResult]): Unit = {
        val elementsOut: util.ArrayList[MobileLog] = new util.ArrayList[MobileLog]()
        elements.foreach(e => elementsOut.add(e))
        var bufferingCountSum: Double = 0
        var assetItemTimeOfSessionSum: Double = 0
        elements foreach (m => {
          bufferingCountSum += (if (m.assetItemTimeOfSession < 18000000 && m.firstVideoTime > 0) m.bufferingCount else 0)
          assetItemTimeOfSessionSum += (if (m.assetItemTimeOfSession > 0 && m.assetItemTimeOfSession < 18000000 && m.firstVideoTime > 0) m.assetItemTimeOfSession / 60000 else 0)
        })
        val value = (if (assetItemTimeOfSessionSum == 0) 0 else bufferingCountSum / assetItemTimeOfSessionSum).formatted("%.6f").toDouble
        out.collect(RateResult(context.window.getStart / 1000, step, metric, value, "", Tags(key, timeCardFrameRate), RateResultFields(bufferingCountSum, assetItemTimeOfSessionSum)))
      }
    }).name(s"移动${business}时长卡顿率").uid("409e80f4-ac06-4b1b-8768-9dc1fa967088" + business)
  }

  /**
   * vv卡顿率<br>
   * count(distinct if(buffering_count>0 and first_video_time > 0,asset_item_session,null))/ <br>
   * count(distinct if(first_video_time > 0, asset_item_session, null) )
   *
   */
  private def vvCardFrameRateCalculate(windowStream: WindowedStream[MobileLog, Tags, TimeWindow], business: String = ""): DataStream[RateResult] = {
    windowStream.process(new ProcessWindowFunction[MobileLog, RateResult, Tags, TimeWindow] {
      override def process(key: Tags, context: Context, elements: Iterable[MobileLog], out: Collector[RateResult]): Unit = {
        val assetItemSessionSet1: mutable.Set[String] = mutable.Set[String]()
        val assetItemSessionSet2: mutable.Set[String] = mutable.Set[String]()
        elements.foreach(m => {
          if (m.bufferingCount > 0 && m.firstVideoTime > 0 && m.assetItemSession != null) assetItemSessionSet1.add(m.assetItemSession)
          if (m.bufferingCount > 0 && m.assetItemSession != null) assetItemSessionSet2.add(m.assetItemSession)
        })
        val value = (if (assetItemSessionSet2.isEmpty) 0 else assetItemSessionSet1.size / assetItemSessionSet2.size.toDouble).formatted("%.6f").toDouble
        out.collect(RateResult(context.window.getStart / 1000, step, metric, value, "", Tags(key, vvCardFrameRate), RateResultFields(assetItemSessionSet1.size, assetItemSessionSet2.size)))
      }
    }).name(s"移动${business} vv卡顿率").uid("e3dc36e9-3c4f-4bba-bed0-b53865e933ba" + business)
  }

  /**
   * 加载失败率<br>
   * 大于5s加载失败率：<br>
   * sum(if(asset_item_time_of_session > 5000 and time_of_video <= 0 and first_video_time = 0, 1, 0 ))/ <br>
   * count(distinct if(first_video_time > 0, asset_item_session, null) ) <br>
   * 大于3s加载失败率：<br>
   * sum(if(asset_item_time_of_session > 3000 and time_of_video <= 0 and first_video_time = 0, 1, 0 ))/ <br>
   * count(distinct if(first_video_time > 0, asset_item_session, null) )<br>
   *
   */
  private def loadFailedRateCalculate(windowStream: WindowedStream[MobileLog, Tags, TimeWindow], business: String = ""): DataStream[RateResult] = {
    windowStream.process(new ProcessWindowFunction[MobileLog, RateResult, Tags, TimeWindow] {
      override def process(key: Tags, context: Context, elements: Iterable[MobileLog], out: Collector[RateResult]): Unit = {
        var count5: Double = 0
        var count3: Double = 0
        val assetItemSessionSet: mutable.Set[String] = mutable.Set[String]()
        elements.foreach(m => {
          if (m.assetItemTimeOfSession > 5000 && m.timeOfVideo <= 0 && m.firstVideoTime == 0) count5 += 1
          if (m.assetItemTimeOfSession > 3000 && m.timeOfVideo <= 0 && m.firstVideoTime == 0) count3 += 1
          if (m.firstVideoTime > 0 && m.assetItemSession != null) assetItemSessionSet.add(m.assetItemSession)
        })
        val value5: Double = (if (assetItemSessionSet.isEmpty) 0 else count5 / assetItemSessionSet.size.toDouble).formatted("%.6f").toDouble
        val value3: Double = (if (assetItemSessionSet.isEmpty) 0 else count3 / assetItemSessionSet.size.toDouble).formatted("%.6f").toDouble
        out.collect(RateResult(context.window.getStart / 1000, step, metric, value5, "", Tags(key, loadFailedRate5), RateResultFields(count5, assetItemSessionSet.size)))
        out.collect(RateResult(context.window.getStart / 1000, step, metric, value3, "", Tags(key, loadFailedRate3), RateResultFields(count3, assetItemSessionSet.size)))
      }
    }).name(s"移动${business}加载失败率（5、3）").uid("91699a80-16a4-4211-aa87-8cbfec57918e" + business)
  }


}
